#include "../include/thread_pool.h"
#include "../include/my_log.h"
#include "../include/macro.h"
#include "../include/global.h"
#include <cerrno>

#include <stdio.h>
//初始化
thread_pool::thread_pool()
{
    if(pthread_mutex_init(&this->mtx,NULL)!=0){
        my_log * log=my_log::get_log_instance();
        log->write_log(LOG_ALERT,errno,"线程池，互斥量初始化失败！！！");
        return;
    }
    if(pthread_cond_init(&this->cond_t,NULL)!=0){
        my_log * log=my_log::get_log_instance();
        log->write_log(LOG_ALERT,errno,"线程池，条件变量初始化失败！！！");
        return;
    }
    this->is_stop_thread_flag=false;
    running_thread_num=0;
}

//销毁
thread_pool::~thread_pool()
{
   
    //销毁
    pthread_mutex_destroy(&this->mtx);
    pthread_cond_destroy(&this->cond_t);


}

void thread_pool::end_all_thread()
{
    this->is_stop_thread_flag=1;    //结束标志
    pthread_cond_broadcast(&cond_t);
    for(int i=0;i<this->thread_vector.size();++i){
        pthread_join(thread_vector[i]->handler,nullptr);
    }
    
    //消息队列里可能还有数据，释放一下----这里应该不会执行，因为数据已经确保处理完了
    while(!message_list.empty()){
        my_log * log=my_log::get_log_instance();
        log->write_log(LOG_ERR,0,"我已经确保了线程池结束前将全部数据处理完，怎么还会到这里 ！！！");
        char *buf=message_list.front();
        message_list.pop_front();
        delete[] buf;
    }

}

//创建线程
bool thread_pool::create_pool(int num)
{
    //创建线程
    thread_item * item;
    for(int i=0;i<num;++i){
        this->thread_vector.push_back(item=new thread_item(this));
        if(pthread_create(&item->handler,NULL,thread_pool::thread_func,item)!=0){       //这里是值传递，传递的是item指向的内存地址，就算后续发生改变，也没关系
            my_log* log=my_log::get_log_instance();
            log->write_log(LOG_ALERT,errno,"创建线程失败");
            return false;
        }

    }
    //这里需要确保一下全部线程都执行了
    for(int i=0;i<thread_vector.size();++i){
        if(thread_vector[i]->if_running==false){
            usleep(100*1000);   //微秒   100*1000=100毫秒
            --i;    //回退一下
        }
    }
    printf("数据处理线程启动完毕 数量为%d\n",num);
    my_log * log=my_log::get_log_instance();
    log->write_log(LOG_INFO,0,"成功初始化线程池");

    return true;
}

//往消息队列里加东西
bool thread_pool::add_message_to_list(char *buf)
{
    //上锁，写入list,通知一个线程
    pthread_mutex_lock(&mtx);   //上锁

    this->message_list.push_back(buf);
    pthread_mutex_unlock(&mtx); //解锁
    //通知一个线程就好了
    if(pthread_cond_signal(&this->cond_t)!=0){
        //没有线程了，说明都在执行，但是加入进去了，写个日志就好了
        my_log * log=my_log::get_log_instance();
        log->write_log(LOG_NOTICE,0,"有任务了，但是加入后，通知线程失败，没有空闲进程");
        return false;
    }
    return true;
}


//线程体函数
void *thread_pool::thread_func(void *thread)
{
    thread_item * item=(thread_item *)(thread);
    thread_pool * pool=item->pool;
    while(1){

        pthread_mutex_lock(&pool->mtx);   //上锁

        while(pool->message_list.size()==0 &&  pool->is_stop_thread_flag==0){   //没有数据，并且程序没有要结束的意思，进入等待
            if(item->if_running==false){
                item->if_running=true;
            }
            
            pthread_cond_wait(&pool->cond_t,&pool->mtx);

        }

        if(pool->is_stop_thread_flag&& pool->message_list.empty()){
            //有退出标志，并需要保证当前没有需要处理的数据
            pthread_mutex_unlock(&pool->mtx);
            break;  //要退出了
        }
        char * buf=pool->message_list.front();
        pool->message_list.pop_front();
        pthread_mutex_unlock(&pool->mtx);   //解锁
        ++pool->running_thread_num;
        //拿到数据了，进行处理吧  调用大类里消息处理函数
        work_socket.thread_for_deal_message(buf);

        //该条消息放在这里delete掉----对于有些需要延迟处理的消息，不能直接销毁
        //delete[]  buf;

        --pool->running_thread_num;
        

        
    }


    return NULL;
}
